package com.zengde.ai_sandbox.controller;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.time.Instant;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;

@RestController
public class StockController {

    private static final int TOTAL_TOKENS = 500;
    private static final ObjectMapper objectMapper = new ObjectMapper();
    private final Random random = new Random();
    private final String[] symbols = {"AAPL", "GOOGL", "MSFT", "AMZN"};

    // 模拟实时股票数据流
    @GetMapping(value = "/stocks", produces = "text/event-stream;charset=UTF-8")
    public Flux<String> streamStocks() {
        final Random random = new Random();
        return Flux.interval(Duration.ofSeconds(1))  // 每秒推送一次
                .map(sequence -> {
                    int index = (int) (sequence % symbols.length);
                    String symbol = symbols[index];
                    // 生成随机股价 (150-200区间波动)
                    double price = 150 + random.nextDouble() * 50;
                    // 返回SSE格式数据（前端可直接处理）
                    return String.format("data: %s 价格: %.2f\n\n", symbol, price);
                });
    }

    @GetMapping(value = "/stocks3", produces = "text/event-stream;charset=UTF-8")
    public Flux<String> streamStocks2() {

//        Flux<String> objectFlux = Flux.create(fluxSink -> {
//            this.streamStocks().subscribe(fluxSink::next);
//            this.streamStocks().subscribe(fluxSink::next,fluxSink::error,fluxSink::complete);
//        });
//
//        return objectFlux;


        return this.streamStocks();


    }


    // 带实时进度的股票流
    @GetMapping(value = "/stocks2", produces = "text/event-stream")
    public Flux<String> stocksWithProgress() {
        final AtomicInteger accumulate = new AtomicInteger(0);

        // 直接调用原始股票流
        return this.streamStocks()
                .map(originalMessage -> {
                    // 1. 提取原始数据内容（去掉SSE格式）
                    String originalContent = originalMessage.substring(6, originalMessage.length() - 2);

                    // 2. 计算当前token长度
                    int tokenCount = originalContent.length();
                    int newTotal = accumulate.addAndGet(tokenCount);

                    // 3. 计算当前进度百分比
                    double percentage = (double) newTotal / TOTAL_TOKENS * 100;

                    // 4. 解析原始股票数据
                    String[] parts = originalContent.split(" ");
                    String symbol = parts[0];
                    double price = Double.parseDouble(parts[2]);

                    // 5. 构建新格式的JSON数据
                    ObjectNode json = objectMapper.createObjectNode();
                    json.put("symbol", symbol);
                    json.put("price", price);
                    json.put("accumulate", newTotal);
                    json.put("total", TOTAL_TOKENS);
                    json.put("percentage", percentage);

                    // 6. 返回新格式的SSE消息
                    return "data: " + json.toString() + "\n\n";
                })
                // 添加完成事件
                .concatWith(Mono.defer(() -> {
                    return Mono.just("event: completion\ndata: 所有数据已发送完成!\n\n");
                }))
                // 总量控制
                .takeWhile(s -> accumulate.get() < TOTAL_TOKENS);
    }


    /**
     * 这是一个错误的示范，不能使用subscribe嵌套，否则会造成资源泄露
     *
     * @return
     */
    @GetMapping(value = "/myRealReactorWrong", produces = "text/event-stream")
    public Flux<String> myRealReactorWrong() {
        Flux<String> flux = Flux.create(fluxSink -> {
            this.streamStocks().subscribe(
                    data -> {
                        System.out.println(data);
                        fluxSink.next(data);
                    },
                    error -> {
                        System.out.println(error);
                        fluxSink.error(error);
                    },
                    () -> {
                        System.out.println("Completed");
                        fluxSink.complete();
                    }
            );
        });
        return flux;
    }


    /**
     * 这个方法用了副作用操作符，也就是doOnComplete,doOnError,doOnNext
     *
     * @return
     */
    @GetMapping(value = "/myRealReactorCorrect", produces = "text/event-stream")
    public Flux<String> myRealReactorCorrent() {
        Flux<String> stringFlux = this.streamStocks()
                .doOnNext(data -> {
                    System.out.println("这个老B的数据为:" + data);
                })
                .doOnComplete(() -> {
                    System.out.println("Completed");
                })
                .doOnError(error -> {
                    System.out.println(error);
                });
        return stringFlux;
    }


    @GetMapping(value = "/myRealcWithMap", produces = "text/event-stream")
    public Flux<String> myRealcWithMap() {
        Flux<String> stringFlux = null;
        stringFlux = this.streamStocks()
                .map(this::processData);
        return stringFlux;
    }


    private String processData(String original) {
        return original + " | 处理时间: " + Instant.now();
    }



    @GetMapping(value = "/myRcAdvanced", produces = "text/event-stream")
    public Flux<String> myRcAdvanced() {
        return this.streamStocks()
                .transform(flux -> Flux.create(sink -> {
                    Disposable disposable = flux.subscribe(
                            data -> {
                                System.out.println("收到你妈B的数据：" + data);
                                sink.next(data);
                            },
                            error -> {
                                System.out.println(error);
                            },
                            () -> {
                                System.out.println("Completed");
                            }
                    );
                    sink.onDispose(disposable::dispose);
                    sink.onCancel(disposable::dispose);
                }));
    }


}





